#include <malloc.h>
#include "string.h"
#include "../includes/mqtt.h"

void (*tcp_send)(char *data_buf, u16 len);

u8 (*tcp_connect)(char *host, u16 port);


MQTT_CONFIG *mqtt_config;
//与服务器通信状态，每秒+1，如果大于keep alive 说明通信有问题
u16 CONNECT_STATUS = 0;

//包ID，自动增减
u16 packet_id = 1;

//发送缓存
u8 send_buf[SEND_BUF_MAX] = {0};
//topic订阅 topic, qos
MQTT_SUBSCRIBE subscribe_buf [SUB_TOPIC_MAX]= {0};
//应答ack，按bit位复用2个byte，高8位结果，低8响应
// 参考.h define ACK_FLAG_BIT_x
volatile u16 FLAG_ACK = 0;

//发送缓存数据位指针
u8 send_buf_point = 0;
//发布标志
u8 publish_flag = false;
//time_tick 时间计数器
u16 time_tick = 0;
u16 last_send = 0;

void clear_buf(char *buf, u16 start, u16 size){
    while (--size){
        buf[start++] = 0;
    }
}

void str_copy(u8 *target, const u8 *source, u8 start, u16 len){
    u16 i = 0; u8 t = 0;
    do {
        target[i++] = source[++start];
    } while (i < len);
}

/**
 * 比较时间过去了多久
 * @param compare
 * @return
 */
u16 get_time_interval(u16 compare){
    if(compare > time_tick){
        return 0xffff - compare + time_tick;
    } else{
        return time_tick - compare;
    }
}

/**
 * 获取剩余长度需要几个字节
 * @param rlen 长度
 * @return 需要字节数
 */
u8 remain_len_size(u16 rlen){
    u8 byte_num = rlen / 127;
    if (rlen % 127 > 0) {
        byte_num++;
    }
    return byte_num > 0 ? byte_num : 1;
}

/**
 * 剩余长度编码
 * @param buf 缓冲区
 * @param start 缓冲区开始位
 * @param rlen 长
 * @return 占用字节数
 */
u8 remain_len_encode(u8 *buf, u16 rlen){
    u8 i = 0, byte_num = remain_len_size(rlen);
    buf[byte_num] = 0x7f & (rlen % 127);
    for (i = 1; i < byte_num; i++) {
        buf[i] = 0xff;
    }
    return byte_num;
}

/**
 * 剩余长度解码
 * @param buf mqtt固定头，接收缓冲区
 * @return 剩余长度
 */
u16 remain_len_decode(u8 *buf) {
    u8 len = 0, i = 1; u16 bs = 1;
    do {
        len += (buf[i] & 0x7f) * bs;
        bs *= 128; //移位乘方
    } while ((buf[i++] & 0x80) != false);
    //取最后得值
    return len;
}

/**
 * 填充utf8字符串
 * @param buf 数组
 * @param start 起始位
 * @param str 字符串
 * @param str_len 字符串长度
 * @return 编码长度
 */
u8 push_str(u8 *buf, u8 start, char *str, u8 str_len) {
    buf[++start] = str_len >> 8;
    buf[++start] = str_len & 0xff;
    str_len = 2;
    while (*str != '\0') {
        buf[++start] = *str;
        str++;
        str_len++;
    }
    return str_len;
}

/**
 * 位操作，将指定位置0/1
 * @param ack_bit 应答标记位
 * @param flag 0/1
 */
void ack_flag_set(u8 ack_bit, u8 flag){
    if(flag){
        FLAG_ACK |= 1 << ack_bit;
    } else{
        FLAG_ACK &= ~ (1 << ack_bit);
    }
}

/**
 * 阻塞式等待标记位重置
 * @param ack_bit bit位
 * @param timeout 等待超时时间 S秒
 * @return 成功还是超时 1=成功 0=超时失败
 */
u8 ack_waiting(u8 ack_bit, u8 timeout){
    u8 temp_time = time_tick;
    while ((FLAG_ACK >> ack_bit) & 0x01){
        if(get_time_interval(temp_time) > timeout){
            ack_flag_set(ack_bit, false);
            return false;
        }
    }
    return true;
}

/**
 * 构造固定头
 * @param type 数据包类型
 * @param rlen 后续数据长度
 * @return 固定头长度
 */
u8 make_fixed_header(PACKET_TYPE type, u8 flag, u16 rlen) {
    if (type == PUBREL || type == SUBSCRIBE || type == UNSUBSCRIBE) {
        send_buf[0] = (type << 4) | 0x02;
    } else {
        send_buf[0] = (type << 4) | flag;
    }
    //计算剩余长度需要多少个字节,为节省内存用发送缓冲当变量
    return remain_len_encode(send_buf, rlen);
}

/**
 * 构造连接包可变头
 * @return
 */
u8 make_CONNECT_var_header() {
    u8 temp = 0;
    send_buf[SEND_BUF_MAX - 1] = 0x00;
    send_buf[SEND_BUF_MAX - 2] = 0x04;
    send_buf[SEND_BUF_MAX - 3] = 'M';
    send_buf[SEND_BUF_MAX - 4] = 'Q';
    send_buf[SEND_BUF_MAX - 5] = 'T';
    send_buf[SEND_BUF_MAX - 6] = 'T';
    send_buf[SEND_BUF_MAX - 7] = VERSION;
    if (mqtt_config->username || strlen(mqtt_config->username) > 0) {
        temp |= 1 << 7;
    } else {
        mqtt_config->username = "";
    }
    if (mqtt_config->password || strlen(mqtt_config->password) > 0) {
        temp |= 1 << 6;
    } else {
        mqtt_config->password = "";
    }
    if (mqtt_config->clean_session == true) {
        temp |= 1 << 1;
    }
    send_buf[SEND_BUF_MAX - 8] = temp;
    send_buf[SEND_BUF_MAX - 9] = mqtt_config->keep_alive >> 8;
    send_buf[SEND_BUF_MAX - 10] = mqtt_config->keep_alive & 0xff;
    return 10;
}

/**
 * 构造发布消息头
 * @param topic
 * @param qos
 * @return
 */
u8 make_PUBLISH_var_header(char *topic, u8 qos) {
    u16 topic_len = strlen(topic);
    send_buf_point += push_str(send_buf, send_buf_point, topic, topic_len);
    if(qos > 0){
        packet_id++;
        send_buf[++send_buf_point] = 0;
        send_buf[++send_buf_point] = 2;
        send_buf[++send_buf_point] = topic_len >> 8;
        send_buf[++send_buf_point] = topic_len & 0x0f;
        return topic_len + 4;
    }
    return topic_len + 2;
}

/**
 * 合并固定头与可变头部
 * @param start_point
 * @param var_len
 */
void merge_header(u8 start_point, u8 var_len) {
    u8 i = 0;
    for (i = 1; i <= var_len; i++) {
        send_buf[start_point + i] = send_buf[SEND_BUF_MAX - i];
    }
    send_buf_point += var_len;
}

/**
 * 处理连接回调
 * @param result 结果，返回状态码
 */
void connect_callback_handler(u8 result){
    switch (result) {
        case 0: mqtt_config -> callback.on_connect(); break;
        case 1: mqtt_config -> callback.on_error("version nonsupport");
        case 2: mqtt_config -> callback.on_error("id be refused");
        case 3: mqtt_config -> callback.on_error("service unavailable");
        case 4: mqtt_config -> callback.on_error("name/psw error");
        case 5: mqtt_config -> callback.on_error("unauthorized");
        default: mqtt_config -> callback.on_error("unknown code");
    }
}

void subscribe_callback_handler(char *buf, u8 start){
    switch (buf[start + 3]) {
        case 0x00:
        case 0x01:
        case 0x02:
            ack_flag_set(FLAG_BIT_ACK_SUBSCRIBE, false);
            break;
        case 0x80:
            //订阅失败
            ack_flag_set(FLAG_BIT_ACK_SUBSCRIBE, false);
            break;
    }
}

void on_msg_callback_handler(char *buf, u8 start, u16 remain_len){
    //取topic
    u16 len = buf[start], msg_len = 0;
    len = (len << 8) | buf[++start];
    msg_len = remain_len - len - 2;
    //直接取topic
    u8 *topic = (u8 *)calloc(sizeof(u8), len);
    str_copy(topic, buf, start, len);
    //取QOS，判断是否为0
    if(((buf[0] >> 1) & 0x03) > 0){
        //取包ID
        len = buf[start + len];
        len = (len << 8) | buf[++start + len];
        u8 *pack_id = (u8 *)calloc(sizeof(u8), len);
        msg_len = remain_len - len - 2;
        //处理包，等待ack应答
        free(pack_id);
    }
    start += len;
    u8 *msg = (u8 *)calloc(sizeof(u8), msg_len);
    str_copy(msg, buf, start, msg_len);
    mqtt_config->callback.on_message(topic, msg);
    free(topic);
    free(msg);
}

/**
 * mqtt 消息解码，服务器返回数据解码
 * @param buf 接收缓存README
 * @param len 接收字节长度
 */
void mqtt_decode(u8 *buf, u16 len) {
    //解码剩余长度
    u16 remain_len = remain_len_decode(buf);
    //可变头部位置指针
    u8 pos = remain_len_size(remain_len);
    //重置连接状态
    CONNECT_STATUS = 0;
    switch (buf[0] >> 4) {
        case CONNACK:
            ack_flag_set(FLAG_BIT_ACK_CONNECT, false);
            if(remain_len >= 2){
                connect_callback_handler(buf[pos + 1]);
            } else {
                mqtt_config -> callback.on_error("decode error");
            }
            break;
        case PUBLISH: //收到订阅的消息
            on_msg_callback_handler(buf, pos + 1, remain_len);
        case PUBACK:
            ack_flag_set(FLAG_BIT_ACK_PUBLISH, false);
            break;
        case PUBREL:
            ack_flag_set(FLAG_BIT_ACK_PUBREL, false);
            break;
        case SUBACK:
            subscribe_callback_handler(buf, pos + 2);
            break;
        case UNSUBACK:
            ack_flag_set(FLAG_BIT_ACK_UNSUBSCRIBE, false);
            break;
        case PINGRESP: break;
    }
}

/**
 * 发起mqtt连接
 * @param config mqtt配置
 * @param net_tcp_send tcp发送函数
 * @param net_tcp_connect tcp连接函数
 */
bool mqtt_connect(MQTT_CONFIG *config, void (*net_tcp_send)(char *data_buf, u16 len),
                  u8 (*net_tcp_connect)(char *host, u16 port)) {
    tcp_send = net_tcp_send;
    tcp_connect = net_tcp_connect;
    mqtt_config = config;
    u8 remain_len = 0, cid_len = 0, usn_len = 0, psw_len = 0;
    remain_len = make_CONNECT_var_header();
    cid_len = strlen(mqtt_config->client_id);
    usn_len = strlen(mqtt_config->username);
    psw_len = strlen(mqtt_config->password);
    send_buf_point = make_fixed_header(CONNECT, FLAG_CONNECT, remain_len
                                            + (cid_len > 0 ? cid_len + 2 : cid_len)
                                            + (usn_len > 0 ? usn_len + 2 : usn_len)
                                            + (psw_len > 0 ? psw_len + 2 : psw_len));
    merge_header(send_buf_point, remain_len);
    //填充payload
    send_buf_point += push_str(send_buf, send_buf_point, mqtt_config->client_id, cid_len);
    if (usn_len > 0) {
        send_buf_point += push_str(send_buf, send_buf_point, mqtt_config->username, usn_len);
    }
    if (psw_len > 0) {
        send_buf_point += push_str(send_buf, send_buf_point, mqtt_config->password, psw_len);
    }
    tcp_connect(config->host, config->port);
    last_send = time_tick;
    ack_flag_set(FLAG_BIT_ACK_CONNECT, true);
    tcp_send(send_buf, send_buf_point + 1);
    return ack_waiting(FLAG_BIT_ACK_CONNECT, ACK_TIMEOUT);
}

/**
 * 发布消息到topic
 * @param topic
 * @param topic_len topic长度
 * @param msg 载荷消息
 * @param len 消息长度
 * @param qos 服务质量
 * @param retain 是否为保留消息
 */
void mqtt_publish(char *topic, u8 topic_len, char *msg, u16 len, u8 qos, u8 retain){
    clear_buf(send_buf, 0, SEND_BUF_MAX);
    u8 rlen = topic_len + 2 + (qos > 0 ? 4 : 0) + len, dup_flag = 0;
    u16 i = 0;
    send_buf_point = make_fixed_header(PUBLISH, (qos << 1 | retain | dup_flag << 3), rlen);
    make_PUBLISH_var_header(topic, qos);
    for (i = 0; i < len; i++){
        send_buf[++send_buf_point] = msg[i];
    }
    //发布标志置0，等待结果返回
    publish_flag = qos > 0 ? 0 : 1;
    last_send = time_tick;
    tcp_send(send_buf, send_buf_point + 1);
}

/**
 * 发布自动长度的消息
 * @param topic
 * @param msg
 * @param qos
 * @param retain
 */
void mqtt_publish_len(char *topic, char *msg, u8 qos, u8 retain){
    mqtt_publish(topic, strlen(topic), msg, strlen(msg), qos, retain);
}

bool mqtt_subscribe(char *topic, u8 qos){
    last_send = time_tick;
    clear_buf(send_buf, 0, SEND_BUF_MAX);
    u8 topic_len = strlen(topic);
    send_buf_point = make_fixed_header(SUBSCRIBE, FLAG_SUBSCRIBE, 5 + topic_len);
    //可变头部
    send_buf[++send_buf_point] = ++packet_id >> 8;
    send_buf[++send_buf_point] = packet_id & 0xff;
    //载荷 topic_len topic qos
    send_buf_point += push_str(send_buf, send_buf_point, topic, topic_len);
    send_buf[++send_buf_point] = qos;
    ack_flag_set(FLAG_BIT_ACK_SUBSCRIBE, true);
    tcp_send(send_buf, send_buf_point + 1);
    //阻塞式等待
    return ack_waiting(FLAG_BIT_ACK_SUBSCRIBE, ACK_TIMEOUT);
}

bool mqtt_unsubscribe(char *topic){
    last_send = time_tick;
    clear_buf(send_buf, 0, SEND_BUF_MAX);
    send_buf_point = make_fixed_header(UNSUBSCRIBE, FLAG_UNSUBSCRIBE, 0);

    tcp_send(send_buf, send_buf_point + 1);
    ack_flag_set(FLAG_BIT_ACK_UNSUBSCRIBE, true);
    //阻塞式等待
    return ack_waiting(FLAG_BIT_ACK_UNSUBSCRIBE, 2);
}

void mqtt_disconnect(){
    //发送disconnect包
    last_send = time_tick;
    clear_buf(send_buf, 0, SEND_BUF_MAX);
    send_buf_point = make_fixed_header(DISCONNECT, FLAG_DISCONNECT, 0);
    tcp_send(send_buf, send_buf_point + 1);
    //断开连接
    mqtt_config->callback.disconnect();
}

void mqtt_ping(){
    last_send = time_tick;
    clear_buf(send_buf, 0, SEND_BUF_MAX);
    send_buf_point = make_fixed_header(PINGREQ, FLAG_PINGREQ, 0);
    tcp_send(send_buf, send_buf_point + 1);
}

void mqtt_sys_tick_s(void){
    //检查keep alive时间，判断是否要ping
    if(mqtt_config -> keep_alive > 0){
        //检查是否要发送ping
        if(get_time_interval(last_send) >= mqtt_config -> keep_alive - 1){
            mqtt_ping();
        }
        CONNECT_STATUS ++;
        //检查是否通信超时
        if(CONNECT_STATUS > mqtt_config -> keep_alive){
            mqtt_config -> callback.disconnect();
        }
    }
    if(++time_tick >= 0xffff){
        time_tick = 0;
    }
}
